package com.zzj.flinkcdcmysqldemo.listener;

import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MysqlDataChangeSink extends RichSinkFunction<String> {

    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void invoke(String value, Context context) {
        log.info("从mysql收到变更数据:{}", value);
        //转换后发送到对应的MQ
        kafkaTemplate.send("flink-cdc-mysql", value);
    }

    /**
     * 在启动SpringBoot项目是加载了Spring容器，其他地方可以使用@Autowired获取Spring容器中的类；但是Flink启动的项目中，
     * 默认启动了多线程执行相关代码，导致在其他线程无法获取Spring容器，只有在Spring所在的线程才能使用@Autowired，
     * 故在Flink自定义的Sink的open()方法中初始化Spring容器
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        kafkaTemplate = SpringUtil.getBean(KafkaTemplate.class);
    }
}
